package com.lagou.hbase.processor;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.HTableWrapper;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

/**
 * 删除rowkey中user时，将其他friends中含有的此user同步删除
 * @author shenxiang
 * @date 2020/9/4 14:45
 * @description
 */
public class UserFriendsDeleteProcessor extends BaseRegionObserver {
    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        HTableWrapper user = (HTableWrapper)e.getEnvironment().getTable(TableName.valueOf("user"));
        byte[] row = delete.getRow();
        List<Cell> friends = delete.getFamilyCellMap().get(Bytes.toBytes("friends"));

        for (Cell cell : friends) {
            byte[] bytes = CellUtil.cloneQualifier(cell);
            Delete delete1 = new Delete(bytes);
            delete1.addColumn(Bytes.toBytes("friends"),row);
            user.delete(delete1);
        }
        user.flushCommits();
        user.close();
    }
}
